package socketmvc.server.tcp;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.core.ChannelContext;
import org.tio.core.Tio;
import org.tio.core.TioConfig;
import org.tio.core.exception.TioDecodeException;
import org.tio.core.intf.Packet;
import org.tio.core.stat.ChannelStat;
import org.tio.core.stat.IpStat;
import org.tio.core.task.DecodeRunnable;
import org.tio.core.utils.ByteBufferUtils;
import org.tio.utils.SystemTimer;
import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
import java.util.concurrent.Executor;

public class TcpDecodeRunnable extends DecodeRunnable {
    private static final Logger log = LoggerFactory.getLogger(TcpDecodeRunnable.class);
    private ChannelContext channelContext = null;
    private TioConfig tioConfig = null;
    /**
     * 上一次解码剩下的数据
     */
    private ByteBuffer lastByteBuffer = null;
    /**
     * 新收到的数据
     */
    private ByteBuffer newReceivedByteBuffer = null;

    /**
     * @param packet
     * @param byteCount
     * @author tanyaowu
     */
    public void handler(Packet packet, int byteCount) {
        switch (tioConfig.packetHandlerMode) {
            case SINGLE_THREAD:
                channelContext.handlerRunnable.handler(packet);
                break;
            case QUEUE:
                channelContext.handlerRunnable.addMsg(packet);
                channelContext.handlerRunnable.execute();
                break;
            default:
                channelContext.handlerRunnable.handler(packet);
                break;
        }
    }

    /**
     *
     */
    public TcpDecodeRunnable(ChannelContext channelContext, Executor executor) {
        super(channelContext, executor);
        this.channelContext = channelContext;
        this.tioConfig = channelContext.tioConfig;
    }

    /**
     * 清空处理的队列消息
     */
    public void clearMsgQueue() {
        super.clearMsgQueue();
        lastByteBuffer = null;
        newReceivedByteBuffer = null;
    }

    /**
     * @author tanyaowu
     * 2017年3月21日 下午4:26:39
     * @see java.lang.Runnable#run()
     */
    public void decode() {
        ByteBuffer byteBuffer = newReceivedByteBuffer;
        if (lastByteBuffer != null) {
            byteBuffer = ByteBufferUtils.composite(lastByteBuffer, byteBuffer);
            lastByteBuffer = null;
        }

        label_2:
        while (true) {
            try {
                int initPosition = byteBuffer.position();
                int limit = byteBuffer.limit();
                int readableLength = limit - initPosition;
                Packet packet = null;
                if (channelContext.packetNeededLength != null) {
                    if (log.isDebugEnabled()) {
                        log.debug("{}, 解码所需长度:{}", channelContext, channelContext.packetNeededLength);
                    }
                    if (readableLength >= channelContext.packetNeededLength) {
                        packet = tioConfig.getTioHandler().decode(byteBuffer, limit, initPosition, readableLength, channelContext);
                    }
                } else {
                    try {
                        packet = tioConfig.getTioHandler().decode(byteBuffer, limit, initPosition, readableLength, channelContext);
                    } catch (BufferUnderflowException e) {
                        //log.error(e.toString(), e);
                        //数据不够读
                    }
                }

                if (packet == null)// 数据不够，解不了码
                {
                    //					lastByteBuffer = ByteBufferUtils.copy(byteBuffer, initPosition, limit);
                    if (tioConfig.useQueueDecode || (byteBuffer != newReceivedByteBuffer)) {
                        byteBuffer.position(initPosition);
                        byteBuffer.limit(limit);
                        lastByteBuffer = byteBuffer;
                    } else {
                        lastByteBuffer = ByteBufferUtils.copy(byteBuffer, initPosition, limit);
                    }
                    ChannelStat channelStat = channelContext.stat;
                    channelStat.decodeFailCount++;
                    //int len = byteBuffer.limit() - initPosition;
                    if (log.isDebugEnabled()) {
                        log.debug("{} 本次解码失败, 已经连续{}次解码失败，参与解码的数据长度共{}字节", channelContext, channelStat.decodeFailCount, readableLength);
                    }
                    if (channelStat.decodeFailCount > 500){
                        log.error("{} 本次解码失败, 已经连续{}次解码失败，参与解码的数据长度共{}字节,有攻击嫌疑", channelContext, channelStat.decodeFailCount, readableLength);
                        throw new TioDecodeException("连续解码失败次数过多");
                    }
                    return;
                } else //解码成功
                {
                    channelContext.setPacketNeededLength(null);
                    channelContext.stat.latestTimeOfReceivedPacket = SystemTimer.currTime;
                    channelContext.stat.decodeFailCount = 0;

                    int packetSize = byteBuffer.position() - initPosition;
                    packet.setByteCount(packetSize);

                    if (tioConfig.statOn) {
                        tioConfig.groupStat.receivedPackets.incrementAndGet();
                        channelContext.stat.receivedPackets.incrementAndGet();
                    }

                    if (tioConfig.isIpStatEnable()) {
                        try {
                            for (Long v : tioConfig.ipStats.durationList) {
                                IpStat ipStat = tioConfig.ipStats.get(v, channelContext);
                                ipStat.getReceivedPackets().incrementAndGet();
                                tioConfig.getIpStatListener().onAfterDecoded(channelContext, packet, packetSize, ipStat);
                            }
                        } catch (Exception e1) {
                            log.error(packet.logstr(), e1);
                        }
                    }

                    if (tioConfig.getTioListener() != null) {
                        try {
                            tioConfig.getTioListener().onAfterDecoded(channelContext, packet, packetSize);
                        } catch (Throwable e) {
                            log.error(e.toString(), e);
                        }
                    }

                    if (log.isDebugEnabled()) {
                        log.debug("{}, 解包获得一个packet:{}", channelContext, packet.logstr());
                    }

                    handler(packet, packetSize);

                    if (byteBuffer.hasRemaining())//组包后，还剩有数据
                    {
                        if (log.isDebugEnabled()) {
                            log.debug("{},组包后，还剩有数据:{}", channelContext, byteBuffer.remaining());
                        }
                        continue label_2;
                    } else//组包后，数据刚好用完
                    {
                        lastByteBuffer = null;
                        if (log.isDebugEnabled()) {
                            log.debug("{},组包后，数据刚好用完", channelContext);
                        }
                        return;
                    }
                }
            } catch (Throwable e) {
                if (channelContext.logWhenDecodeError) {
                    log.error("解码时遇到异常", e);
                }

                channelContext.setPacketNeededLength(null);

                if (e instanceof TioDecodeException && tioConfig.isIpStatEnable()) {
                    try {
                        for (Long v : tioConfig.ipStats.durationList) {
                            IpStat ipStat = tioConfig.ipStats.get(v, channelContext);
                            ipStat.getDecodeErrorCount().incrementAndGet();
                            tioConfig.getIpStatListener().onDecodeError(channelContext, ipStat);
                        }
                    } catch (Exception e1) {
                        log.error(e1.toString(), e1);
                    }
                }

                Tio.close(channelContext, e, "解码异常:" + e.getMessage(), ChannelContext.CloseCode.DECODE_ERROR);
                return;
            }
        }
    }

    /**
     * @param newReceivedByteBuffer
     */
    public void setNewReceivedByteBuffer(ByteBuffer newReceivedByteBuffer) {
        this.newReceivedByteBuffer = newReceivedByteBuffer;
    }

    @Override
    public String toString() {
        return this.getClass().getSimpleName() + ":" + channelContext.toString();
    }

    @Override
    public String logstr() {
        return toString();
    }

}
